# -*- coding: utf-8 -*-
__author__ = 'sunnychou'
__date__ = '2019/9/16 10:57'

'''
封装数据分析功能
  便于进行报表合并，分组，计算等系列操作
采用基础库：pandas
'''

import  pandas as pd
from pandas import Series, DataFrame
from sqlalchemy  import create_engine
from common.logger_helper import g_wlogger
#数据库连接
MYSQL_CONNECT_STR = "mssql+pymssql://sa:Favor@sqladmin123@180.76.115.157/CRM6?charset=utf8"

ENGINE  = create_engine(MYSQL_CONNECT_STR)

class PandasHelper(object):

    @staticmethod
    def create_dataframe(data=[],  columns=[]):
        '''
        创建dataframe
        :param data:
        :param columns:
        :return:
        '''
        return DataFrame(data=data, columns=columns)

    @staticmethod
    def  pd_query_sql(p_sql):
        '''
        :param p_sql: 查询sql语句
                eg: select   month, gongys_name, bibie, hanshui_unit, jiashui_heji   from   k3_caigoud  where month BETWEEN  '2019-01' and '2019-09'
        :return:  pandas dataframe
        说明：
          一定要将处理计算的字段进行数字转换成float64，才能进行字段计算
          df_result['jiashui_heji'] = pd.to_numeric(df_result['jiashui_heji'])
        '''
        try:
            print(f"p_sql:{p_sql}")
            qr_df = pd.read_sql_query(p_sql, con=ENGINE)
            #qr_df = qr_df.convert_objects(convert_numeric=True)  #转换数字类型
            return qr_df
        except Exception as e:
            g_wlogger.werror(f"pd_query_sql has error {e} with: {p_sql}")
            return pd.DataFrame()

    @staticmethod
    def pd_read_excel(excel_name, dtype='str'):
        '''
        采用pandas读取excel文件
        :param excel_name:  eg: 2019年1月份应收账款分析表.xls
        :param dtype: 读取数据类型，str是字符串类型，float是以浮点数读取
        :return:dataframe
        '''
        try:
            return  pd.read_excel(excel_name, dtype=dtype)
        except:
            return pd.DataFrame()

    @staticmethod
    def df_drop_duplicates(df):
        '''
        dataframe数据去重
        :param df:
        :return:
        '''
        return df.drop_duplicates()


    @staticmethod
    def df_cond_filter(df, conds):
       '''
       :param df:    传入dataframe
       :param conds: 条件
            单条件：  (df['month'] > '2019-06')
            多条件： (df['month'] == '2019-07') & (df['gongys_name'] == '上海斯丹麦德电子有限公司')
       :return: df[(df['month'] == '2019-07') & (df['gongys_name'] == '上海斯丹麦德电子有限公司')]
       '''
       try:
           return df[conds]
       except Exception as e:
           g_wlogger.werror(f"df_cond_filter has error {e} with: {conds}")
           return pd.DataFrame()


    @staticmethod
    def df_query(df, strquery):
        '''
        按照传入的strquery进行dataframe的查询 df.query('gongys_name == ["上海斯丹麦德电子有限公司"]')
        :param df: dataframe数据集
        :param strquery: gongys_name == ["上海斯丹麦德电子有限公司"]
        :return:
        '''
        try:
            query_str = f"{strquery}"
            query_df = df.query(query_str)
            if len(query_df) == 0:
                return pd.DataFrame()
            return query_df
        except Exception as e:
            g_wlogger.werror(f"df_query has error {e} with: {strquery}")
            return pd.DataFrame()   #返回一个空的dataframe，可以通过empty判断是否为空


    @staticmethod
    def  df_cal_field_sum(df, gby_fname, field_name):
        '''
        按照某一个字段进行分组，然后按照另一个字段进行计算(求和)
        :param df:
        :param gby_fname: 分组字段  eg:gongys_name   group_df = result.groupby('gongys_name')
        :param field_name: 待计算字段  jiashui_heji
        :return: series，一维数组
        '''
        try:
            group_df = df.groupby(f"{gby_fname}")
            group_series = group_df[f"{field_name}"].sum()
            return group_series
        except Exception as e:
            g_wlogger.werror(f"df_cal_field_sum has error {e}, gby_fname:{gby_fname}")
            return pd.DataFrame()


    @staticmethod
    def  series_field_value(mseries, key):
        '''
        返回一维数组series对应的key的value
        :param mseries:
        :param key:
        :return:
        '''
        return mseries[key]


    @staticmethod
    def df_field_transform(df, field_name, target_field_name, callfunc):
        '''
        字段转换计算，返回新的字段
        eg: df['打分'] = cresult['month'].map(transform_month)
        :param field_name: 待参考计算字段       eg: month
        :param target_field_name: 目标新增字段, eg：打分
        :param callfunc:  回调函数              eg: transform_month
        :return:
        '''
        #对field_name字段的每一个值都传入callfunc函数计算，返回结果赋值给target_field_name字段
        df[f"'{target_field_name}'"] = df[f"'{field_name}'"].map(callfunc)
        return df


    @staticmethod
    def df_pivot_table(df, index=0, values=0, columns=0, aggfunc=None, margins=False,margins_name='合计',fill_value=0):
        '''
        透视表进行数据集的合并和计算特定字段（求和，求平均值）
        df.pivot_table(index=['gongys_name'], values=['jiashui_heji'], columns=['month'], aggfunc='sum',
                            margins=True,
                            margins_name='金额求和值', fill_value=0)
        :param df:
        :param index:  作为新表的列名，可以是多个（list格式）
        :param values:  要进行汇总、统计等运算的列，可以是多个（list格式）
        :param columns: 作为新表的列名，可以是多个（list格式）
        :param aggfunc: 做汇总的统计运算方法
        :param margins: margins=True来添加行列合计
        :param margins_name: margins_name=' '来修改汇总行列的名称，也可以使用rename()来修改
        :param fill_value: fill_value = 0 是用来填充缺失值、空值
        :return:
        '''
        df_pivot_tb = df.pivot_table(index=index, values=values, columns=columns, aggfunc=aggfunc,
                       margins=margins, margins_name=margins_name, fill_value=fill_value)
        return df_pivot_tb


    @staticmethod
    def df_fetch_iloc_value(df, row_index=-1, columns_index=-1):
        '''
        返回给行索引和列索引对应的位置值
        df.iloc[row_index:row_index + 1, columns_index:columns_index + 1]
        :param df:
        :param row_index:     int, 行索引从0开始，左开右闭 [2:3]  ---> 2
        :param columns_index: int, 列索引从0开始
        :return:
        '''
        if (row_index == -1) and (columns_index == -1):
            return df.iloc[:,:]
        elif (row_index == -1) and (columns_index != -1):
            return df.iloc[:,columns_index:columns_index+1]
        elif (row_index != -1) and (columns_index == -1):
            return df.iloc[row_index:row_index+1, :]
        else:
            #return df.iloc[row_index:row_index+1, columns_index:columns_index+1].values[0][0]
            return df.iloc[row_index:row_index + 1, columns_index:columns_index + 1]


    @staticmethod
    def select_df_some_columns(df, column_list, fill_column_list, fillna_value=0):
        '''
        从df中选择某些列
        :param df: 需要处理的df
        :param column_list:  ['col11','col22','col33']
        :param  fill_column_list  ['mol1', 'mol2', 'mol3']
        :param  fillna_value: 填充nan值为fillna_value
        :return: dataframe
        '''
        assert (len(column_list) != len(fill_column_list))
        select_df =  df[column_list]
        select_df = select_df.fillna(fillna_value)   #填充
        select_df.columns = fill_column_list
        return select_df

    @staticmethod
    def df_to_excel(df, excel_file="test_excel.xls"):
        '''
        将dataframe数据集df直接写入excel
        :param df:
        :param excel_file: excel文件
        :return:
        '''
        try:
            df.to_excel(excel_file, index=False)  # 不写index
            return True
        except Exception as e:
            g_wlogger.werror(f"PandasHelper::df_to_excel has error {e} write with: {excel_file}")
            return False

    @staticmethod
    def  df_to_sql(df, table_name, index=False, if_exists='fail'):
        '''
        写df数据到sql数据库表中
        :param df: 待处理dataframe
        :param table_name: 需要保存的表名称
        :param index: 索引是否保存
        :param if_exists: {'fail', 'replace', 'append'}
                 * fail: Raise a ValueError.
                 * replace: Drop the table before inserting new values.
                 * append: Insert new values to the existing table.
        :return:
        '''
        try:
            df.to_sql(table_name, con=ENGINE, index=index, if_exists=if_exists)
            return True
        except Exception as e:
            g_wlogger.werror(f"df_to_sql has error {e}")
            return False


    @staticmethod
    def fetch_records_from_engine(query_sql):
        '''
        传入sql语句查询记录
        :param query_sql:  SELECT * FROM users where ...
        :return:
        '''
        try:
            return  ENGINE.execute(query_sql).fetchall()
        except Exception as e:
            g_wlogger.werror(f"fetch_records_from_engine has error {e} with query_sql:{query_sql}")
            return []